package org.mq.mymq.broker.netty.nameserver;

import com.alibaba.fastjson2.JSON;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.DelimiterBasedFrameDecoder;
import io.netty.util.internal.StringUtil;
import org.mq.mymq.broker.cache.CommonCache;
import org.mq.mymq.broker.config.GlobalProperties;
import org.mq.mymq.common.coder.TcpMsg;
import org.mq.mymq.common.coder.TcpMsgDecoder;
import org.mq.mymq.common.coder.TcpMsgEncoder;
import org.mq.mymq.common.constants.TcpConstants;
import org.mq.mymq.common.dto.PullBrokerIpDTO;
import org.mq.mymq.common.dto.PullBrokerIpRespDTO;
import org.mq.mymq.common.dto.ServiceRegistryReqDTO;
import org.mq.mymq.common.enums.*;
import org.mq.mymq.common.remote.NameServerNettyRemoteClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.Inet4Address;
import java.net.UnknownHostException;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;

/**
 * @ClassName NameServerClient
 * @Description 负责与nameserver服务端创建长链接，支持链接创建，支持重试机制
 * @Author jiarong_ye
 * @Date 2025/6/19 16:00
 * @Version 1.0
 */
public class NameServerClient {
    private static final Logger log = LoggerFactory.getLogger(NameServerClient.class);
    private EventLoopGroup clientGroup = new NioEventLoopGroup();
    private Bootstrap bootstrap = new Bootstrap();
    private Channel channel;
    private String DEFAULT_USER_SERVER_IP = "127.0.0.1";

    private NameServerNettyRemoteClient nameServerNettyRemoteClient;



    /**
     * 初始化链接
     *
     * @throws InterruptedException
     */
    public void initConnection() {
        String ip = CommonCache.getGlobalProperties().getNameserverIp();
        Integer port = CommonCache.getGlobalProperties().getNameserverPort();
        if (StringUtil.isNullOrEmpty(ip) || port == null || port < 0) {
            throw new RuntimeException("error port or ip");
        }
        nameServerNettyRemoteClient = new NameServerNettyRemoteClient(ip, port);
        nameServerNettyRemoteClient.builldConnection();
    }

    public Channel getChannel() {
        if (channel == null) {
            throw new RuntimeException("TCP connection not established");
        }
        return channel;
    }

    public void sendRegistryMsg() {
        try {
            Map<String, Object> attrs = new HashMap<String, Object>();
            GlobalProperties globalProperties = CommonCache.getGlobalProperties();
            String clusterMode = globalProperties.getBrokerClusterMode();
            if (StringUtil.isNullOrEmpty(clusterMode)) {
                attrs.put("role", "single");
            } else if(BrokerClusterModeEnum.MASTER_SLAVE.getCode().equals(clusterMode)) {
                // 注册模式是集群模式
                BrokerRegistryRoleEnum brokerRegistryRoleEnum = BrokerRegistryRoleEnum.of(globalProperties.getBrokerClusterRole());
                attrs.put("role", brokerRegistryRoleEnum.getCode());
                attrs.put("group", globalProperties.getBrokerClusterGroup());
            }
            ServiceRegistryReqDTO dto = new ServiceRegistryReqDTO();
            dto.setIp(Inet4Address.getLocalHost().getHostAddress());
            dto.setPort(globalProperties.getBrokerPort());
            dto.setUser(globalProperties.getNameserverUser());
            dto.setPassword(globalProperties.getNameserverPassword());
            dto.setRegistryType(RegistryTypeEnum.BROKER.getCode());
            dto.setAttrs(attrs);
            dto.setMsgId(UUID.randomUUID().toString());
            byte[] body = JSON.toJSONBytes(dto);
            TcpMsg tcpMsg = new TcpMsg(NameServerEventCode.REGISTRY.getCode(), body);
            TcpMsg registryResponseMsg = nameServerNettyRemoteClient.sendSyncMsg(tcpMsg, dto.getMsgId());
            if (NameServerResponseCode.REGISTRY_SUCCESS.getCode() == registryResponseMsg.getCode()) {
                // 注册成功，开启定时任务上报心跳数据给nameserver
                log.info("注册成功，开启心跳任务");
                CommonCache.getHeartBeatTaskManager().startTask();
            } else if (NameServerResponseCode.ERROR_USER_OR_PASSWORD.getCode() == registryResponseMsg.getCode()) {
                // 验证失败，抛出异常
                throw new RuntimeException(NameServerResponseCode.ERROR_USER_OR_PASSWORD.getDesc());
            } else if (NameServerResponseCode.HEART_BEAT_SUCCESS.getCode() == registryResponseMsg.getCode()) {
                log.info("收到nameserver心跳回应ack");
            }
            log.info("registry resp is:{}", com.alibaba.fastjson.JSON.toJSONString(registryResponseMsg.getBody()));
        } catch (UnknownHostException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 获取broker主节点的地址
     *
     * @return
     */
    public String queryBrokerMasterAddress() {
        String clusterMode = CommonCache.getGlobalProperties().getBrokerClusterMode();
        if (!BrokerClusterModeEnum.MASTER_SLAVE.getCode().equals(clusterMode)) {
            return null;
        }
        PullBrokerIpDTO pullBrokerIpDTO = new PullBrokerIpDTO();
        pullBrokerIpDTO.setRole(BrokerRegistryRoleEnum.MASTER.getCode());
        pullBrokerIpDTO.setBrokerClusterGroup(CommonCache.getGlobalProperties().getBrokerClusterGroup());
        pullBrokerIpDTO.setMsgId(UUID.randomUUID().toString());
        TcpMsg tcpMsg = new TcpMsg(NameServerEventCode.PULL_BROKER_IP_LIST.getCode(), JSON.toJSONBytes(pullBrokerIpDTO));
        TcpMsg respMsg = nameServerNettyRemoteClient.sendSyncMsg(tcpMsg, pullBrokerIpDTO.getMsgId());
        PullBrokerIpRespDTO pullBrokerIpRespDTO = JSON.parseObject(respMsg.getBody(), PullBrokerIpRespDTO.class);
        return pullBrokerIpRespDTO.getMasterAddressList().get(0);
    }

    public NameServerNettyRemoteClient getNameServerNettyRemoteClient() {
        return nameServerNettyRemoteClient;
    }
}
